//! 事件总线实现
//! 
//! 提供插件间异步通信机制，支持：
//! - 事件发布和订阅
//! - 异步事件处理
//! - 事件过滤和路由
//! - 事件持久化（可选）

use std::collections::HashMap;
use std::sync::Arc;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use tokio::sync::{broadcast, RwLock};
use uuid::Uuid;

use crate::error::{KernelError, Result};

/// 事件类型
#[derive(Debug, Clone, Serialize, Deserialize, Hash, Eq, PartialEq)]
pub enum EventType {
    /// 插件生命周期事件
    PluginLifecycle,
    /// 查询执行事件
    QueryExecution,
    /// 数据源事件
    DataSource,
    /// 配置变更事件
    ConfigChange,
    /// 安全事件
    Security,
    /// 系统事件
    System,
    /// 自定义事件
    Custom(String),
}

/// 事件数据
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Event {
    /// 事件ID
    pub id: String,
    /// 事件类型
    pub event_type: EventType,
    /// 事件源
    pub source: String,
    /// 事件目标（可选）
    pub target: Option<String>,
    /// 事件数据
    pub data: serde_json::Value,
    /// 事件时间戳
    pub timestamp: chrono::DateTime<chrono::Utc>,
    /// 事件元数据
    pub metadata: HashMap<String, String>,
}

impl Event {
    /// 创建新事件
    pub fn new(
        event_type: EventType,
        source: String,
        data: serde_json::Value,
    ) -> Self {
        Self {
            id: Uuid::new_v4().to_string(),
            event_type,
            source,
            target: None,
            data,
            timestamp: chrono::Utc::now(),
            metadata: HashMap::new(),
        }
    }
    
    /// 设置事件目标
    pub fn with_target(mut self, target: String) -> Self {
        self.target = Some(target);
        self
    }
    
    /// 添加元数据
    pub fn with_metadata(mut self, key: String, value: String) -> Self {
        self.metadata.insert(key, value);
        self
    }
    
    /// 获取事件类型
    pub fn event_type(&self) -> &EventType {
        &self.event_type
    }
}

/// 事件处理器接口
#[async_trait]
pub trait EventHandler: Send + Sync {
    /// 处理事件
    async fn handle(&self, event: &Event) -> Result<()>;
    
    /// 获取处理器ID
    fn id(&self) -> &str;
    
    /// 获取感兴趣的事件类型
    fn interested_events(&self) -> Vec<EventType>;
}

/// 事件过滤器
pub type EventFilter = Box<dyn Fn(&Event) -> bool + Send + Sync>;

/// 事件总线
pub struct EventBus {
    /// 事件发送器
    sender: broadcast::Sender<Event>,
    /// 事件处理器
    handlers: Arc<RwLock<HashMap<EventType, Vec<Arc<dyn EventHandler>>>>>,
    /// 事件过滤器
    filters: Arc<RwLock<Vec<EventFilter>>>,
    /// 事件历史（可选）
    history: Arc<RwLock<Vec<Event>>>,
    /// 最大历史记录数
    max_history: usize,
}

impl EventBus {
    /// 创建新的事件总线
    pub fn new() -> Self {
        let (sender, _) = broadcast::channel(1000);
        
        Self {
            sender,
            handlers: Arc::new(RwLock::new(HashMap::new())),
            filters: Arc::new(RwLock::new(Vec::new())),
            history: Arc::new(RwLock::new(Vec::new())),
            max_history: 10000,
        }
    }
    
    /// 创建带配置的事件总线
    pub fn with_config(channel_size: usize, max_history: usize) -> Self {
        let (sender, _) = broadcast::channel(channel_size);
        
        Self {
            sender,
            handlers: Arc::new(RwLock::new(HashMap::new())),
            filters: Arc::new(RwLock::new(Vec::new())),
            history: Arc::new(RwLock::new(Vec::new())),
            max_history,
        }
    }
    
    /// 订阅事件
    pub async fn subscribe(&self, handler: Arc<dyn EventHandler>) -> Result<()> {
        let mut handlers = self.handlers.write().await;
        
        for event_type in handler.interested_events() {
            handlers.entry(event_type)
                .or_insert_with(Vec::new)
                .push(handler.clone());
        }
        
        tracing::debug!("事件处理器已订阅: {}", handler.id());
        Ok(())
    }
    
    /// 取消订阅
    pub async fn unsubscribe(&self, handler_id: &str) -> Result<()> {
        let mut handlers = self.handlers.write().await;
        
        for (_, handler_list) in handlers.iter_mut() {
            handler_list.retain(|h| h.id() != handler_id);
        }
        
        tracing::debug!("事件处理器已取消订阅: {}", handler_id);
        Ok(())
    }
    
    /// 发布事件
    pub async fn publish(&self, event: Event) -> Result<()> {
        // 应用过滤器
        {
            let filters = self.filters.read().await;
            for filter in filters.iter() {
                if !filter(&event) {
                    tracing::debug!("事件被过滤器拒绝: {}", event.id);
                    return Ok(());
                }
            }
        }
        
        // 记录事件历史
        {
            let mut history = self.history.write().await;
            history.push(event.clone());
            
            // 限制历史记录数量
            if history.len() > self.max_history {
                history.remove(0);
            }
        }
        
        // 发送事件到广播通道
        if let Err(e) = self.sender.send(event.clone()) {
            tracing::warn!("发送事件到广播通道失败: {}", e);
        }
        
        // 直接调用相关处理器
        {
            let handlers = self.handlers.read().await;
            if let Some(handler_list) = handlers.get(&event.event_type) {
                for handler in handler_list {
                    if let Err(e) = handler.handle(&event).await {
                        tracing::error!("事件处理器执行失败 {}: {}", handler.id(), e);
                    }
                }
            }
        }
        
        tracing::debug!("事件已发布: {} (类型: {:?})", event.id, event.event_type);
        Ok(())
    }
    
    /// 添加事件过滤器
    pub async fn add_filter(&self, filter: EventFilter) -> Result<()> {
        let mut filters = self.filters.write().await;
        filters.push(filter);
        Ok(())
    }
    
    /// 获取事件历史
    pub async fn get_history(&self, limit: Option<usize>) -> Vec<Event> {
        let history = self.history.read().await;
        
        match limit {
            Some(n) => {
                let start = if history.len() > n { history.len() - n } else { 0 };
                history[start..].to_vec()
            },
            None => history.clone(),
        }
    }
    
    /// 清空事件历史
    pub async fn clear_history(&self) -> Result<()> {
        let mut history = self.history.write().await;
        history.clear();
        Ok(())
    }
    
    /// 创建事件接收器
    pub fn subscribe_channel(&self) -> broadcast::Receiver<Event> {
        self.sender.subscribe()
    }
    
    /// 获取统计信息
    pub async fn get_stats(&self) -> EventBusStats {
        let handlers = self.handlers.read().await;
        let history = self.history.read().await;
        let filters = self.filters.read().await;
        
        let mut handler_count_by_type = HashMap::new();
        for (event_type, handler_list) in handlers.iter() {
            handler_count_by_type.insert(event_type.clone(), handler_list.len());
        }
        
        EventBusStats {
            total_handlers: handlers.values().map(|v| v.len()).sum(),
            handler_count_by_type,
            total_events_processed: history.len(),
            filter_count: filters.len(),
            channel_capacity: 1000, // 这里应该从实际配置获取
        }
    }
}

/// 事件总线统计信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EventBusStats {
    pub total_handlers: usize,
    pub handler_count_by_type: HashMap<EventType, usize>,
    pub total_events_processed: usize,
    pub filter_count: usize,
    pub channel_capacity: usize,
}

#[cfg(test)]
mod tests {
    use super::*;
    
    struct TestEventHandler {
        id: String,
        events: Arc<RwLock<Vec<Event>>>,
    }
    
    impl TestEventHandler {
        fn new(id: String) -> Self {
            Self {
                id,
                events: Arc::new(RwLock::new(Vec::new())),
            }
        }
        
        async fn get_received_events(&self) -> Vec<Event> {
            let events = self.events.read().await;
            events.clone()
        }
    }
    
    #[async_trait]
    impl EventHandler for TestEventHandler {
        async fn handle(&self, event: &Event) -> Result<()> {
            let mut events = self.events.write().await;
            events.push(event.clone());
            Ok(())
        }
        
        fn id(&self) -> &str {
            &self.id
        }
        
        fn interested_events(&self) -> Vec<EventType> {
            vec![EventType::System, EventType::Custom("test".to_string())]
        }
    }
    
    #[tokio::test]
    async fn test_event_bus() {
        let event_bus = EventBus::new();
        
        // 创建测试处理器
        let handler = Arc::new(TestEventHandler::new("test-handler".to_string()));
        
        // 订阅事件
        event_bus.subscribe(handler.clone()).await.unwrap();
        
        // 发布事件
        let event = Event::new(
            EventType::System,
            "test-source".to_string(),
            serde_json::json!({"message": "test"}),
        );
        
        event_bus.publish(event.clone()).await.unwrap();
        
        // 等待事件处理
        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
        
        // 检查事件是否被处理
        let received_events = handler.get_received_events().await;
        assert_eq!(received_events.len(), 1);
        assert_eq!(received_events[0].id, event.id);
        
        // 检查事件历史
        let history = event_bus.get_history(None).await;
        assert_eq!(history.len(), 1);
        
        // 获取统计信息
        let stats = event_bus.get_stats().await;
        assert_eq!(stats.total_events_processed, 1);
        assert!(stats.total_handlers > 0);
    }
    
    #[tokio::test]
    async fn test_event_filter() {
        let event_bus = EventBus::new();
        
        // 添加过滤器，只允许System类型的事件
        event_bus.add_filter(Box::new(|event| {
            matches!(event.event_type, EventType::System)
        })).await.unwrap();
        
        // 发布System事件
        let system_event = Event::new(
            EventType::System,
            "test-source".to_string(),
            serde_json::json!({"message": "system"}),
        );
        event_bus.publish(system_event).await.unwrap();
        
        // 发布Custom事件（应该被过滤）
        let custom_event = Event::new(
            EventType::Custom("test".to_string()),
            "test-source".to_string(),
            serde_json::json!({"message": "custom"}),
        );
        event_bus.publish(custom_event).await.unwrap();
        
        // 检查历史，应该只有一个事件
        let history = event_bus.get_history(None).await;
        assert_eq!(history.len(), 1);
        assert!(matches!(history[0].event_type, EventType::System));
    }
}
